package homomorphism.list;

import homomorphism.HomomorphismBase;
import utility.PairT;
import utility.LongPair;
import java.io.IOException;
import java.util.ArrayList;
import java.lang.reflect.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

/**
 *
 * @author takeshi
 */
public abstract class SecondMapperFor3rdHomo<InVal,OtVal>  extends MapReduceBase implements Mapper<LongPair,InVal,LongPair,OtVal> {

    private static final Log LOG = LogFactory.getLog(SecondMapperFor3rdHomo.class);
    private static final String userAppProperty = HomomorphismBase.UserClassNameProperty;
   // private Class FoldingType;
    private Class UnFoldingType;
    private String UserClassName;

    @Override
    public void configure(JobConf job) {

        super.configure(job);
        UserClassName = job.get(userAppProperty);
     //   FoldingType = job.getMapOutputValueClass();
        UnFoldingType = job.getOutputValueClass() ;
    }

    public ArrayList<OtVal> unfolding(InVal value) throws Exception  {
        ClassLoader myloader = FirstReducerFor3rdHomo.class.getClassLoader();
        Class UserApp;
        if( UserClassName ==null || UserClassName.isEmpty() ) return null;
        try {
            UserApp = myloader.loadClass( UserClassName );
            Class partypes[] = new Class[1];
             partypes[0] =  UnFoldingType;
            Method _unfold = UserApp.getMethod( "unfold", partypes);
           
            Object arglist[] = new Object[1];
            arglist[0] = value;
            Constructor ct = UserApp.getConstructor();//partypes
            Object methobj = ct.newInstance();
            ArrayList<OtVal> rst = (ArrayList<OtVal>) _unfold.invoke(methobj, arglist);
            return rst;

        } catch (ClassNotFoundException ee) {
            LOG.error(ee.getMessage());
        }
        return null;
    }

    public void map(LongPair key, InVal value, OutputCollector<LongPair, OtVal> output, Reporter reporter) throws IOException {

        LongPair newKey = new LongPair(0, key.getSecond());// key(pid,index)
        ArrayList<OtVal> local = null;
        try {
            local = unfolding(value);
        } catch (Exception ex) {
            LOG.error(ex.toString());
        }

        output.collect(newKey, local.get(0));
        LongPair newKey2 = new LongPair(newKey.getFirst(), newKey.getSecond() + 1l);
        output.collect(newKey2, local.get(1));
        local.clear();
    }

}

